package com.fs.iot.collector;

import com.fs.iot.bean.dto.GatewayDTO;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.jetbrains.annotations.NotNull;
import org.springframework.stereotype.Service;

import java.time.Duration;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.regex.Pattern;

/**
 * kafka的实时工况直接处理器
 *
 * @author LiuQi 2025/3/13-9:21
 * @version V1.0
 **/
@Service
@Slf4j
public class KafkaDataCollector extends AbstractDataCollector {
    private static final String GROUP_ID = "qi-iot";

    private static final Map<String, KafkaConsumer<String, String>> clientMap = new Hashtable<>();

    private static final ExecutorService executorService = Executors.newFixedThreadPool(100);

    // 停止监听
    @Override
    public void stopListener(String gatewayId) {
        Optional.ofNullable(clientMap.get(gatewayId)).ifPresent(client -> {
            try {
                client.close();
                clientMap.remove(gatewayId);
            } catch (Exception e) {
                log.error("关闭连接失败", e);
            }
        });
    }

    /**
     * 启动MQTT监听
     */
    @Override
    public void startListener(GatewayDTO gateway) {
        // 需要判断之前的连接是否存在
        String gatewayId = gateway.getId();
        this.stopListener(gatewayId);

        KafkaConsumer<String, String> consumer = getConsumer(gateway);
        consumer.subscribe(Pattern.compile(".*"));
        clientMap.put(gatewayId, consumer);
        while (clientMap.containsKey(gatewayId)) {
            ConsumerRecords<String, String> record = consumer.poll(Duration.ofSeconds(30));
            record.forEach(r -> {
                String value = r.value();
                String topic = r.topic();
                executorService.submit(() -> this.process(gateway, topic, value));
            });
        }
    }

    /**
     * 获取数据采集协议
     *
     * @return 协议
     */
    @Override
    protected String getProtocol() {
        return "kafka";
    }

    @NotNull
    private static KafkaConsumer<String, String> getConsumer(GatewayDTO gateway) {
        Properties props = new Properties();
        String host = gateway.getConnectionInfo().getHost();
        String port = gateway.getConnectionInfo().getPort();
        props.setProperty("bootstrap.servers", host + ":" + port);
        props.setProperty("group.id", GROUP_ID);
        props.setProperty("enable.auto.commit", "true");
        props.setProperty("auto.commit.interval.ms", "1000");
        props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        return new KafkaConsumer<>(props);
    }
}
